#include "./include/rpcprovider.h"
#include"zookeeperutil.h"
#define test



/*

为了远程rpc请求过来定位本地方法
需要根据下面1，2，3，4组织下述结构
service_name->service描述
    service* 记录服务对象
    method_name => method方法对象

*/

void RpcProvider::NotifyService(google::protobuf::Service* service){
    //Service提供哪些方法描述服务对象和服务方法
    ServiceInfo service_info;
    service_info.m_service = service;
    //1.获取服务对象的描述信息
    const google::protobuf::ServiceDescriptor* pserviceDesc = service->GetDescriptor();
    //2.获取服务的名字
    std::string service_name = pserviceDesc->name();

    // std::cout<<"service_name: "<<service_name<<std::endl;
    LOG_INFO("service_name: %s",service_name.c_str());



    //3.获取服务对象service的方法的数量
    int methodcnt = pserviceDesc->method_count();
#ifdef test
    // std::cout<<"methodcnt: "<<methodcnt<<std::endl;
    LOG_INFO("methodcnt: %d",methodcnt);
#endif  

    for(int i =0;i<methodcnt;++i){
        //4.获取服务对象的某一服务方法的描述( )
        const google::protobuf::MethodDescriptor* pmethodDesc  =  pserviceDesc->method(i);
        service_info.m_mthodMap.insert(std::make_pair(pmethodDesc->name(),pmethodDesc));
#ifdef test
        // std::cout<<"methodname: "<<pmethodDesc->name()<<std::endl;
        LOG_INFO("methodname: %s",pmethodDesc->name().c_str());
#endif  
    }
    m_serviceMap.insert(std::make_pair(service_name,service_info));

}

void RpcProvider::Run(){
    std::string rpcserverip = MprpcApplication::GetInstance().GetConfig().Load("rpcserverip");
    uint16_t rpcserverport =stoi(MprpcApplication::GetInstance().GetConfig().Load("rpcserverport"));
    muduo::net::InetAddress rpcserveraddress(rpcserverip,rpcserverport);

    //创建一个rpcproviderServer
    muduo::net::TcpServer server(&m_eventLoop,rpcserveraddress,"RpcProvidere");

    //绑定连接回调
    server.setConnectionCallback(std::bind(&RpcProvider::OnConnection,this,std::placeholders::_1));

    //绑定消息读写回调
    server.setMessageCallback(std::bind(&RpcProvider::OnMessage,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
    
    //设置muduo库的线程数量
    server.setThreadNum(4); 

    //把当前rpc节点上要发布的服务全部注册到zk上面，让rpc_client可以从zk上发现服务
    // session timeout==>30s zkclient中的网络I/O线程会一直已1/3*timeout时间发生ping消息
    ZkClient zkcli;
    zkcli.Start(); //连接zookeeperserver
    //将已经发布到rpc节点的服务注册到zkserver上面
    //service_name为永久性节点 method_name为临时节点
    for(auto& sp:m_serviceMap){
        std::string service_path = "/"+sp.first;
        //必须先创建父节点 
        zkcli.Create(service_path.c_str(),nullptr,0);
        for(auto& mp:sp.second.m_mthodMap){
            //service_name/method_name
            std::string method_path = service_path+"/"+mp.first;
            char method_path_data[128] = {0};
            sprintf(method_path_data,"%s:%d",rpcserverip.c_str(),rpcserverport);
            //ZOO_EPHEMERAL是一个临时性节点
            //ZOO_SEQUENCE是一个永久性节点
            zkcli.Create(method_path.c_str(),method_path_data,sizeof(method_path_data),ZOO_EPHEMERAL);
        }
    }

    //提示信息
    // std::cout<<"RpcProvider start service at ip: "<<rpcserverip<<" port: "<<rpcserverport<<std::endl;
    LOG_INFO("RpcProvider start service at ip:  %s   port: %d  ",rpcserverip.c_str(),rpcserverport);

    

    //启动网络服务
    server.start();
    m_eventLoop.loop(); //启动epoll_wait()

} 


void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr&conn){
    //rpc服务是短连接，对端关闭后，本方也关闭
    if(!conn->connected()){
        conn->shutdown();
    }
}

 /*
 接收的buffer是用户提供的相关请求信息
 在框架内部，Rpcprovider和RpcConsumer协商好之间通信用的protobuf数据类型
 service_name method_name args 定义proto的message类型，进行数据头的序列化和反序列化
 
 
 */

 
//请求的响应，序列化和反序列化在这执行
void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer,muduo::Timestamp){
     //网络上接收的远程rpc调用请求的字符流 
    //buffer是Message中rpcheader的转化存储信息
    std::string recv_buf =  buffer->retrieveAllAsString();  //这个是二进制字符串

    //从字符流中读取前4个字节的内容
    uint32_t head_size = 0;
    recv_buf.copy((char*)&head_size,4);

    //根据接收到的headsize读取数据头的原始字符流，反序列化数据，得到rpc请求得详细信息
    std::string rpc_header_str = recv_buf.substr(4,head_size);
    mprpc::Rpcheader rpcheader;
    std::string service_name;
    std::string method_name;
    uint32_t args_size;
    if(rpcheader.ParseFromString(rpc_header_str)){
        //数据头反序列化成功
        service_name = rpcheader.serivice_name();
        method_name = rpcheader.method_name();
        args_size = rpcheader.args_size();
    }
    else{
        //数据反序列化失败
        // std::cout<<"rpc_header_str: "<<rpc_header_str<<"parse error!"<<std::endl;
        LOG_ERR("rpc_header_str: %s ,parse error! %s %s %d",rpc_header_str.c_str(),__FILE__,__FUNCTION__,__LINE__);
        return;
    }

    //获取rpc方法参数得字符流数据
    std::string args_str = recv_buf.substr(4+head_size,args_size);
    
#ifdef test
    // std::cout<<"=============================================="<<std::endl;
    // std::cout<<"head_size: "<<head_size<<std::endl;
    LOG_INFO("head_size: %d",head_size);
    // std::cout<<"rpc_header_str: "<<rpc_header_str<<std::endl;
    LOG_INFO("rpc_header_str: %s",rpc_header_str.c_str());
    // std::cout<<"service name: "<<service_name<<std::endl;
    LOG_INFO("service name: %s",service_name.c_str());
    // std::cout<<"method_name: "<<method_name<<std::endl;
    LOG_INFO("method_name: %s",method_name.c_str());
    // std::cout<<"args_str: "<<args_str<<std::endl;
    LOG_INFO("arg_str: %s",args_str.c_str());
    // std::cout<<"=============================================="<<std::endl;

#endif

    //获取service对象和method对象
    if(m_serviceMap.find(service_name)==m_serviceMap.end()){
        // std::cout<<service_name<<" is not exist!"<<std::endl;
        LOG_ERR("%s is not exist! %s %s %d",service_name.c_str(),__FILE__,__FUNCTION__,__LINE__);
        return;
    }
    

    if(m_serviceMap[service_name].m_mthodMap.find(method_name)==m_serviceMap[service_name].m_mthodMap.end()){
        // std::cout<<service_name<<": "<<method_name<<" is not exist!"<<std::endl;
        LOG_ERR("%s : %s is not exist! %s %s %d",service_name.c_str(),method_name.c_str(),__FILE__,__FUNCTION__,__LINE__);
        return;
    }

    google::protobuf::Service* service = m_serviceMap[service_name].m_service;   //获取service对象 
    const google::protobuf::MethodDescriptor* method = m_serviceMap[service_name].m_mthodMap[method_name]; //获取method对象

    //生成rpc方法调用请求得request和响应response参数（callee里面使用protobuf生成得login中得request 和response）
    google::protobuf::Message* request =  service->GetRequestPrototype(method).New();
    
    if(!request->ParseFromString(args_str)){
        // std::cout<<"request parse error, content: "<<args_str<<std::endl;
        LOG_ERR("request parse error, content: %s %s %s %d",args_str.c_str(),__FILE__,__FUNCTION__,__LINE__);
        return;
    }
    google::protobuf::Message* response = service->GetResponsePrototype(method).New();
    
    //为了不用自己去定义一个派生类，继承自 ::google::protobuf::Closure，重写里面得run方法，protobuf给我们提供了一个API
    //Newcallback生成了一个对象，为其中的run绑定了一个回调函数
    google::protobuf::Closure* done = google::protobuf::NewCallback<RpcProvider,const muduo::net::TcpConnectionPtr &,google::protobuf::Message*>(this,&RpcProvider::SendRpcResponse,conn,response);
    
    
    //在框架上根据远端rpc请求，调用对应的方法
    service->CallMethod(method,nullptr,request,response,done);

      
    
     

}

//Closure的回调操作，用于序列化rpcresponse
void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr &conn,google::protobuf::Message* response){
    //done->run拟执行的方法
    //1.序列化response
    std::string response_str;
    //response进行序列化
    if(response->SerializeToString(&response_str)){
        //序列化成功后，通过网络把rpc方法执行的结果发送回调用法
        conn->send(response_str);
        //rpc按道理是短链接，所以给客户端回送以后，rpc_server主动断开(模拟http的短链接服务)
        conn->shutdown();
    }
    else{
        // std::cout<<"serialize response_str error!"<<std::endl;
        LOG_ERR("serialize response_str error! %s %s %d",__FILE__,__FUNCTION__,__LINE__);
    }
}



